package com.innodealing.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import java.util.HashMap;
import java.util.Map;

/**
 * 自定义 Kafka 消费者配置
 * 演示通过配置文件指定 groupId 的方式
 */
@Configuration
public class SingleManualConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers:localhost:9092}")
    private String bootstrapServers;

    @Value("${kafka.consumer.order-group.group-id:single-manual-example-group}")
    private String orderGroupId;

    @Value("${kafka.consumer.order-group.auto-offset-reset:latest}")
    private String autoOffsetReset;

    /**
     * 创建订单处理专用的消费者工厂
     */
    @Bean("singleManualConsumerFactory")
    public ConsumerFactory<String, String> singleManualConsumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, orderGroupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);

        // 手动提交模式
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 10000);

        return new DefaultKafkaConsumerFactory<>(props);
    }

    /**
     * 创建订单处理专用的监听器容器工厂
     */
    @Bean("singleManualKafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> singleManualKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();

        factory.setConsumerFactory(singleManualConsumerFactory());

        // 设置确认模式为手动立即确认
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

        // 配置重试模板 - 3次重试，每次间隔2秒
        RetryTemplate retryTemplate = new RetryTemplate();

        // 设置重试策略：一直重试
        retryTemplate.setRetryPolicy(new SimpleRetryPolicy(Integer.MAX_VALUE));

        // 设置固定退避策略：每次重试间隔2秒
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(2000L); // 2秒
        retryTemplate.setBackOffPolicy(backOffPolicy);

        factory.setRetryTemplate(retryTemplate);

        // 重试耗尽时的恢复回调
        factory.setRecoveryCallback(context -> {
            System.err.println("处理重试耗尽: " + context.getLastThrowable().getMessage());
            return null;
        });

        // 设置并发数
        factory.setConcurrency(2);

        return factory;
    }
}
